package org.ala.tiktools.rpc.tcp.client_server;

import java.util.concurrent.CompletableFuture;

import org.ala.tiktools.rpc.tcp.config.RpcTcpConfigInfo;

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.WriteBufferWaterMark;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

/**
 * netty相关工具
 *
 * @author ala
 * @date 2025-04-08 12:37
 */
public class TcpNettyTools {


    /************************************************************************************************
     *  服务端相关
     ************************************************************************************************/
    /**
     *  服务端的ServiceBootstrap
     */
    static ServerBootstrap serverBootstrap;
    /**
     *  构建服务端BootStrap
     */
    public static ServerBootstrap serverBootstrap(int bossCount, int workerCount, ChannelInitializer<Channel> initializer) {
        if (serverBootstrap == null) {
            synchronized (TcpNettyTools.class) {
                if (serverBootstrap == null) {
                    serverBootstrap = new ServerBootstrap();
                    NioEventLoopGroup bosser = new NioEventLoopGroup(bossCount);
                    NioEventLoopGroup worker = new NioEventLoopGroup(workerCount);
                    serverBootstrap.childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, RpcTcpConfigInfo.INSTANCE.getConnectionTimeoutMills())
                            //  缓冲区相关
                            .childOption(ChannelOption.SO_SNDBUF, RpcTcpConfigInfo.INSTANCE.getSoSndbuf())
                            .childOption(ChannelOption.SO_RCVBUF, RpcTcpConfigInfo.INSTANCE.getSoRcvbuf())
                            .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(
                                    RpcTcpConfigInfo.INSTANCE.getSoWriteBufferLowWaterMark(),
                                    RpcTcpConfigInfo.INSTANCE.getSoWriteBufferHighWaterMark()))
                            //  tcp相关
                            .childOption(ChannelOption.SO_KEEPALIVE, RpcTcpConfigInfo.INSTANCE.isSoKeepalive())
                            .childOption(ChannelOption.TCP_NODELAY, RpcTcpConfigInfo.INSTANCE.isTcpNoDelay())
                            .childOption(ChannelOption.SO_LINGER, RpcTcpConfigInfo.INSTANCE.getSoLinger())
                            //  读写控制相关
                            .childOption(ChannelOption.AUTO_READ, RpcTcpConfigInfo.INSTANCE.isAutoRead())
                            .childOption(ChannelOption.AUTO_CLOSE, RpcTcpConfigInfo.INSTANCE.isAutoClose())
                            //  其他
                            .childOption(ChannelOption.SO_REUSEADDR, RpcTcpConfigInfo.INSTANCE.isSoReUseAddr())
                            .childOption(ChannelOption.ALLOW_HALF_CLOSURE, RpcTcpConfigInfo.INSTANCE.isAllowHalfClosure())
                            .childOption(ChannelOption.WRITE_SPIN_COUNT, RpcTcpConfigInfo.INSTANCE.getWriteSpinCount());
                    serverBootstrap.group(bosser, worker)
                            .channel(NioServerSocketChannel.class)
                            .childHandler(initializer)
                    ;
                }
            }
        }
        return serverBootstrap;
    }
    /**
     *  启动服务端
     */
    public static Channel bind(int bossCount, int workerCount, int port, ChannelInitializer<Channel> initializer) {
        ChannelFuture cf = serverBootstrap(bossCount, workerCount, initializer).bind(port);
        Channel channel = null;
        //  同步开启
        try {
            channel = cf.sync().channel();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return channel;
    }




    /************************************************************************************************
     *  客户端相关
     ************************************************************************************************/
    /**
     *  客户端统一Bootstrap
     */
    static Bootstrap bootstrap = null;
    /**
     *  构建客户端BootStrap
     */
    public static Bootstrap bootstrap(int workerCount, ChannelInitializer<Channel> initializer) {
        if (bootstrap == null) {
            synchronized (TcpNettyTools.class) {
                if (bootstrap == null) {
                    bootstrap = new Bootstrap();
                    NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(workerCount);
                    bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, RpcTcpConfigInfo.INSTANCE.getConnectionTimeoutMills())
                            //  缓冲区相关
                            .option(ChannelOption.SO_SNDBUF, RpcTcpConfigInfo.INSTANCE.getSoSndbuf())
                            .option(ChannelOption.SO_RCVBUF, RpcTcpConfigInfo.INSTANCE.getSoRcvbuf())
                            .option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(
                                    RpcTcpConfigInfo.INSTANCE.getSoWriteBufferLowWaterMark(),
                                    RpcTcpConfigInfo.INSTANCE.getSoWriteBufferHighWaterMark()))
                            //  tcp相关
                            .option(ChannelOption.SO_KEEPALIVE, RpcTcpConfigInfo.INSTANCE.isSoKeepalive())
                            .option(ChannelOption.TCP_NODELAY, RpcTcpConfigInfo.INSTANCE.isTcpNoDelay())
                            .option(ChannelOption.SO_LINGER, RpcTcpConfigInfo.INSTANCE.getSoLinger())
                            //  读写控制相关
                            .option(ChannelOption.AUTO_READ, RpcTcpConfigInfo.INSTANCE.isAutoRead())
                            .option(ChannelOption.AUTO_CLOSE, RpcTcpConfigInfo.INSTANCE.isAutoClose())
                            //  其他
                            .option(ChannelOption.SO_REUSEADDR, RpcTcpConfigInfo.INSTANCE.isSoReUseAddr())
                            .option(ChannelOption.ALLOW_HALF_CLOSURE, RpcTcpConfigInfo.INSTANCE.isAllowHalfClosure())
                            .option(ChannelOption.WRITE_SPIN_COUNT, RpcTcpConfigInfo.INSTANCE.getWriteSpinCount());
                    bootstrap.group(eventLoopGroup)
                            .channel(NioSocketChannel.class)
                            .handler(initializer)
                    ;
                }
            }
        }
        return bootstrap;
    }
    /**
     *  建立链接（异步的）
     */
    public static CompletableFuture<Channel> connect(String host, int port, ChannelInitializer<Channel> initializer) {
        ChannelFuture cf = bootstrap(RpcTcpConfigInfo.INSTANCE.getClientWorkerCount(), initializer).connect(host, port);
        CompletableFuture<Channel> rtn = new CompletableFuture<Channel>();
        cf.addListener(f -> {
            if (f.isSuccess() && !rtn.isCancelled()) {
                rtn.complete((SocketChannel)cf.channel());
            } else {
                rtn.completeExceptionally(f.cause());
            }
        });
        return rtn;
    }
}
